-module(rtps_writer_proxy).

%%% @doc The writer proxy is kept by a stateful reader to keep track of
%%% the remote writer. The writer proxy keeps track of which messages
%%% were received previously and in case of being reliable, will
%%% request resending data from the remote writer.
%%%
%%% TODO: Include Heartbeat_suppression_duration

-behaviour(gen_statem).

-include_lib("kernel/include/logger.hrl").
-include("rtps.hrl").
-include("rtps_cfw.hrl").
-include("rtps_history_cache.hrl").

%% API
-export([start_link/5, get_qos/1]).

%% gen_statem callbacks
-export([init/1, callback_mode/0,
	 handle_event/4, 
	 terminate/3, code_change/4]).

-record(state, {reliability_level :: reliable | best_effort,
		state1 :: waiting | may_send_ack | must_send_ack}).
%% See table 8.55
-record(data, {remote_writer_guid :: rtps:guid(), reader_guid :: rtps:guid(),
	       reader_cache :: pid(), expected_seq_num :: rtps:sequence_number() | undefined,
	       cfws :: [rtpc_cfw:cfw()] | undefined, % changes_from_writer
	       expects_inline_qos :: boolean(), qos :: rtps_qos:qos() | undefined,
	       acknack_count :: pos_integer() | undefined,
	       heartbeat_response_delay :: rtps:duration() | undefined,
	       heartbeat_count :: non_neg_integer() | undefined,
	       %% heartbeat_suppression_duration :: rtps:duration() | undefined,
	       transports :: undefined | [pid()]}).

%%%===================================================================
%%% API
%%%===================================================================

start_link(Remote_writer_guid, Reg, Reader_guid, Reader_cache, Opts) ->
    gen_statem:start_link({via, rtps_reg, {Reg, Remote_writer_guid}}, ?MODULE, [Remote_writer_guid, Reader_guid, Reader_cache, Opts], []).

-spec get_qos(Pid) -> Qos | undefined when
    Pid :: pid(),
    Qos :: [rtps:parameter()].
%% @doc This operation allows access to the existing set of QoS
%% policies for the writer_proxy.
get_qos(Pid) ->
    gen_statem:call(Pid, get_qos).

%%%===================================================================
%%% gen_statem callbacks
%%%===================================================================

%% @private
init([Remote_writer_guid, Reader_guid, Reader_cache, Opts]) ->
    Reliability_level = proplists:get_value(reliability_level, Opts),
    %% Connect to the transport. This will return the pid of the
    %% sender process associated with transport.
    Transports_sup = proplists:get_value(transports_sup, Opts),
    U_locs = proplists:get_value(uniicast_locator_list, Opts, []),
    M_locs = proplists:get_value(multicast_locator_list, Opts), % always has a default multicast locator
    Locs = U_locs ++ M_locs,
    Transports = [begin 
		      {ok, Pid} = rtps_transport_sup:add(Transports_sup, self(), Locator,
							 Reader_guid, Remote_writer_guid, Opts),
		      Pid
		  end
		  || Locator <- Locs],
    Expects_inline_qos = proplists:get_value(expects_inline_qos, Opts, true),
    Qos = case Expects_inline_qos of
	      true -> [];
	      false -> undefined
	  end,
    Data = #data{remote_writer_guid = Remote_writer_guid, reader_guid = Reader_guid,
		 reader_cache = Reader_cache, expects_inline_qos = Expects_inline_qos, qos = Qos},
    case Reliability_level of
	best_effort ->
	    case rtps_history_cache:get_seq_num_max(Reader_cache, Remote_writer_guid) of
		undefined ->
		    Expected_seq_num = 1;
		Avail ->
		    Expected_seq_num = Avail + 1
	    end,
	    {ok, #state{reliability_level = best_effort, state1 = waiting},
	     Data#data{expected_seq_num = Expected_seq_num}};
	reliable ->
	    Cfws = case rtps_history_cache:get_seq_num_min_max(Reader_cache, Remote_writer_guid) of
		       undefined ->
			   rtps_cfw:new();

		       {Min_seq_num, Max_seq_num} ->
			   rtps_cfw:new(Min_seq_num, Max_seq_num)
		   end,
	    %% Table 8.62 - RTPS Reader configuration attributes
	    Heartbeat_response_delay = proplists:get_value(heartbeat_response_delay, Opts, ?HEARTBEAT_RESPONSE_DELAY), % ms
	    {ok, #state{reliability_level = reliable, state1 = waiting},
	     Data#data{cfws = Cfws,
		       heartbeat_response_delay = Heartbeat_response_delay,
		       heartbeat_count = 0, acknack_count = 1, transports = Transports}}
    end.

%% @private
callback_mode() ->
    handle_event_function.

%% @private

%% ----- API Events

handle_event({call, From}, get_qos, _State, #data{qos = Qos}) ->
    %% ?LOG_DEBUG("Unhandled event: Type ~p, Content ~p", [Event_type,  Event_content]),
    Reply = Qos,
    {keep_state_and_data, [{reply, From, Reply}]};

%% ----- Events from the receiver

%% 8.4.12.1.2 Transition T2 DATA message is received from the matched
%% Writer.
%% 
%% TODO: the Src_guid is NOT used as the writer's guid in the cache
%% change, even though it is the same as the remote writer's guid and
%% used in the specs as such. What about the situation that a reader
%% must store cache changes on behalf og another reader?
handle_event(info, {rcv, Rcvs},
	     #state{reliability_level = best_effort},
	     #data{remote_writer_guid = Remote_writer_guid, reader_cache = Reader_cache,
		   expected_seq_num = Expected_seq_num,
		   expects_inline_qos = Expects_inline_qos, qos = Qos} = Data) ->
    %% Filter out the relevant data submessages
    %% TODO: add cache_change kind?!
    %% TODO: Should we check on every sequence number being higher than the previous one?
    Fun1 = fun(#msg{submsg = #data_msg{writer_sn = Seq_num, serialized_data = Serialized_data,
				       serialized_key = Serialized_key, inline_qos = Inline_qos},
		    timestamp = Timestamp, size = Size})
		 when Seq_num >= Expected_seq_num ->
		   Cache_change = #cache_change{sequence_number = Seq_num,
						writer_guid = Remote_writer_guid,
						data_value = Serialized_data,
						instance_handle = Serialized_key,
						timestamp = Timestamp,
						size = Size},
		   {true, {Cache_change, Inline_qos}};
	      (_Rcv) ->
		   false
	   end,
    {Cache_changes, Inline_qoss} = lists:unzip(lists:filtermap(Fun1, Rcvs)),
    rtps_history_cache:add_change(Reader_cache, Cache_changes),
    Qos1 = case Expects_inline_qos of
	       true ->
		   merge_inline_qos([Qos | Inline_qoss]);
	       false ->
		   Qos
    end,
    #cache_change{sequence_number = Last_seq_num} = lists:last(Cache_changes),
    {keep_state, Data#data{expected_seq_num = Last_seq_num + 1, qos = Qos1}};

handle_event(info, {rcv, Rcvs}, _State, _Data) ->
    {keep_state_and_data, [{next_event, internal, Rcvs}]};
handle_event(internal, [], _State, _Data) ->
    keep_state_and_data;

%% 8.4.12.2.2 Transition T2 and 8.4.12.2.7 Transition T7 Cont'd TODO:
%% the specs only specify the handling of HEARTBEATs if the state is
%% waiting, but it might be possible that heartbeats arrive while the
%% state is must_send_ack, which may result in a different
%% acknack. Currently, we just stick to the specs and ignore
%% HEARTBEATS while in the must_send state.
handle_event(internal,
	     [#msg{submsg = #heartbeat{first_sn = First_sn, last_sn = Last_sn, 
				       count = Count, final_flag = Final_flag,
				       liveliness_flag = Liveliness_flag}} | Rcvs],
	     #state{reliability_level = reliable, state1 = State1} = State,
	     #data{cfws = Cfws,
		   heartbeat_response_delay = Heartbeat_response_delay,
		   heartbeat_count = Heartbeat_count} = Data)
  when Count > Heartbeat_count ->
    Cfws1 = rtps_cfw:missing_changes_update(Cfws, Last_sn),
    Cfws2 = rtps_cfw:lost_changes_update(Cfws1, First_sn),
    Data1 = Data#data{cfws = Cfws2, heartbeat_count = Count + 1},
    %% See Figure 8.24 for details
    case {Final_flag, Liveliness_flag} of
	%% T2 HEARTBEAT message is received, HB.FinalFlag == NOT_SET
	{false, _} when State1 =:= waiting ->
	    {next_state, State#state{state1 = must_send_ack}, Data1,
	     [{{timeout, heartbeat_response_delay}, Heartbeat_response_delay, undefined},
	      {next_event, internal, Rcvs}]};
	%% T2 HEARTBEAT message is received, HB.FinalFlag == SET and HB.LivelinessFlag == NOT_SET
	%% NB: The whole may_send_ack status is redundant
	{true, false} when State1 =:= waiting ->
	    case rtps_cfw:missing_changes(Cfws2) of
		[]  ->
		    {next_state, State#state{state1 = waiting}, Data1,
		     [{next_event, internal, Rcvs}]};
		_ ->
		    {next_state, State#state{state1 = must_send_ack}, Data1,
		     [{{timeout, heartbeat_response_delay}, Heartbeat_response_delay, undefined},
		      {next_event, internal, Rcvs}]}
	    end;
	%% T2 HEARTBEAT message is received, HB:FinalFlag == SET and HB:LivelinessFlag == SET
	{true, true} when State1 =:= waiting ->
	    {keep_state, Data1, [{next_event, internal, Rcvs}]};
	%% Any other condition. The clause above is redundant but kept for clarity.
	_ ->
	    {keep_state, Data1, [{next_event, internal, Rcvs}]}
    end;
%% HEARTBEATs with the wrong count are dropped.
handle_event(internal, [#msg{submsg = #heartbeat{}} | Rcvs],
	     #state{reliability_level = reliable} = _State,
	     _Data) ->
    {keep_state_and_data, [{next_event, internal, Rcvs}]};

%% 8.4.12.2.8 Transition T8 DATA message is received.
handle_event(internal, 
	     [#msg{submsg = #data_msg{writer_sn = Seq_num, serialized_data = Serialized_data,
				      serialized_key = Serialized_key, inline_qos = Inline_qos},
		   timestamp = Timestamp, size = Size} | Rcvs],
	     #state{reliability_level = reliable},
	     #data{cfws = Cfws, remote_writer_guid = Remote_writer_guid,
		   expects_inline_qos = Expects_inline_qos, qos = Qos,
		   reader_cache = Reader_cache} = Data) ->
    Change = #cache_change{sequence_number = Seq_num, writer_guid = Remote_writer_guid,
			   data_value = Serialized_data, instance_handle = Serialized_key,
			   timestamp = Timestamp, size = Size},
    Qos1 = case Expects_inline_qos of
	       true ->
		   merge_inline_qos([Qos | [Inline_qos]]);
	       false ->
		   Qos
    end,
    case rtps_history_cache:add_change(Reader_cache, Change) of
	ok ->
	    Cfws1 = rtps_cfw:received_change_set(Cfws, Seq_num),
	    {keep_state, Data#data{cfws = Cfws1, qos = Qos1}, [{next_event, internal, Rcvs}]};
	{error, already_present} ->
	    {keep_state_and_data, [{next_event, internal, Rcvs}]}
    end;

%% 8.4.12.2.9 Transition T9 GAP message is received.
handle_event(internal,
	     [#msg{submsg = #gap{gap_start = Gap_start, gap_list = [Base | _] = Gap_list}} | Rcvs],
	     #state{reliability_level = reliable},
	     #data{cfws = Cfws} = Data) ->
    Cfws1 = irrelevant_change_set(Cfws, lists:seq(Gap_start, Base - 1)),
    Cfws2 = irrelevant_change_set(Cfws1, Gap_list),
    {keep_state, Data#data{cfws = Cfws2}, [{next_event, internal, Rcvs}]};

%% ----- Time-out Events

%% 8.4.12.2.5 Transition T5 after(R::heartbeatResponseDelay). 
handle_event({timeout, heartbeat_response_delay}, _Content,
	     #state{reliability_level = reliable, state1 = must_send_ack} = State,
	     #data{cfws = Cfws, remote_writer_guid = Remote_writer_guid, reader_guid = Reader_guid,
		   transports = Transports, acknack_count = Count} = Data) ->
    Avail_seq_num = rtps_cfw:available_changes_max(Cfws),
    Missing_seq_nums = rtps_cfw:missing_changes(Cfws),
    %% What value should the Final flag have in the ACKNACK?
    Final_flag = true,
    Msg = rtps_submsg:acknack(Reader_guid, Remote_writer_guid, Avail_seq_num,
			    Missing_seq_nums, Count, Final_flag),
    
    %% Send the ACKNACK on all available transports
    [rtps_sender:send(Pid, Msg) || Pid <- Transports],
    {next_state, State#state{state1 = waiting}, Data#data{acknack_count = Count + 1}};
handle_event({timeout, heartbeat_response_delay}, _Content, _State, _Data) ->
    keep_state_and_data;

%% TODO: Sink unhadled events. Remove these clauses.
handle_event({call, From} = _Event_type, _Event_content, _State, _Data) ->
    Reply = ok,
    {keep_state_and_data, [{reply, From, Reply}]};
handle_event(_Event_type,  _Event_content, _State, _Data) ->
    keep_state_and_data.

%% @private
terminate(_Reason, _State, _Data) ->
    ok.

%% @private
code_change(_Old_vsn, State, Data, _Extra) ->
    {ok, State, Data}.

%%%===================================================================
%%% Internal functions
%%%===================================================================

irrelevant_change_set(Cfws, []) ->
    Cfws;
irrelevant_change_set(Cfws, [Seq_num | Rest]) ->
    rtps_cfw:irrelevant_change_set(Cfws, Seq_num),
    irrelevant_change_set(Cfws, Rest).

merge_inline_qos(Inline_qoss) ->
    merge_inline_qos(Inline_qoss, []).

merge_inline_qos([], Acc) ->
    Acc;
merge_inline_qos([L | Rest], Acc) ->
    L1 = lists:keysort(#parameter.id, L),
    Acc1 = lists:keymerge(#parameter.id, Acc, L1),
    merge_inline_qos(Rest, Acc1).
